package io.github.quickmsg.interate;

import cn.hutool.core.util.ObjectUtil;
import io.github.quickmsg.common.integrate.Integrate;
import io.github.quickmsg.common.integrate.cluster.IntegrateCluster;
import io.github.quickmsg.common.message.mqtt.ClusterMessage;
import io.github.quickmsg.common.utils.ServerUtils;
import io.github.quickmsg.interate.sharedstrategy.SharedStrategy;
import io.github.quickmsg.interate.sharedstrategy.SharedStrategyFactory;
import io.github.quickmsg.interate.sharedstrategy.SharedStrategyType;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.ImmutableTriple;
import org.apache.ignite.IgniteMessaging;
import org.apache.ignite.cluster.ClusterGroup;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.lang.IgniteBiPredicate;

import java.io.Serializable;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @author luxurong
 */
public class IgniteIntegrateCluster implements IntegrateCluster, Serializable {

    private final IgniteIntegrate igniteIntegrate;

    private final IgniteMessaging message;

    private final org.apache.ignite.IgniteCluster igniteCluster;

    private final Map<String, IgniteBiPredicate<UUID, ?>> fixedListener = new ConcurrentHashMap<>();


    private final ClusterHandler clusterHandler;

    public IgniteIntegrateCluster(IgniteIntegrate igniteIntegrate) {
        this.igniteIntegrate = igniteIntegrate;
        this.message = igniteIntegrate.getIgnite().message();
        this.igniteCluster = igniteIntegrate.getIgnite().cluster();
        this.clusterHandler = new ClusterHandler();
    }


    @Override
    public Set<String> getClusterNode() {
        return igniteCluster
                .nodes()
                .stream()
                .map(clusterNode -> clusterNode.consistentId().toString())
                .collect(Collectors.toSet());
    }

    @Override
    public Set<String> getOtherClusterNode() {
        return igniteCluster
                .nodes()
                .stream()
                .filter(clusterNode -> clusterNode != igniteCluster.localNode())
                .map(clusterNode -> clusterNode.consistentId().toString())
                .collect(Collectors.toSet());
    }

    @Override
    public String getLocalNode() {
        return igniteIntegrate.getIgnite().cluster().localNode().addresses().stream().findFirst().orElse(ServerUtils.serverIp);
    }


    @Override
    public void listenTopic(String topic) {
        fixedListener.computeIfAbsent(topic, tp -> {
                    IgniteBiPredicate<UUID, ?> p = clusterHandler::doRemote;
                    message.localListen(tp, p);
                    return p;
                }
        );
    }

    @Override
    public void stopListenTopic(String topic) {
        Optional.ofNullable(fixedListener.remove(topic))
                .ifPresent(o -> message.stopLocalListen(topic, o));
    }

    @Override
    public void sendCluster(String topic, ClusterMessage clusterMessage) {
        message.send(topic, clusterMessage);
    }

    @Override
    public void sendSharedSubscribe(String topic, ClusterMessage clusterMessage) {
        //如果topic有共享订阅，就采用共享订阅模式发送
        Set<ImmutableTriple<String, String, String>> queueSubscribeTopics = this.igniteIntegrate.getTopics().getQueueSubscribeTopics(topic);
        if (CollectionUtils.isNotEmpty(queueSubscribeTopics)) {
            sendSharedMessage(clusterMessage, queueSubscribeTopics, SharedStrategyType.round_robin);
        }
        //如果topic有分组订阅，就采取分组订阅模式发送
        Set<ImmutableTriple<String, String, String>> shareSubscribeTopics = this.igniteIntegrate.getTopics().getShareSubscribeTopics(topic);
        if (CollectionUtils.isNotEmpty(shareSubscribeTopics)) {
            Map<String, Set<ImmutableTriple<String, String, String>>> shareSubscribeMap = new ConcurrentHashMap<>();
            for (ImmutableTriple<String, String, String> tr : shareSubscribeTopics) {
                shareSubscribeMap.computeIfAbsent(tr.middle, s -> new HashSet<>()).add(tr);
            }
            for (Map.Entry<String, Set<ImmutableTriple<String, String, String>>> entry : shareSubscribeMap.entrySet()) {
                sendSharedMessage(clusterMessage, entry.getValue(), SharedStrategyType.round_robin);
            }
        }
    }

    private void sendSharedMessage(ClusterMessage clusterMessage, Set<ImmutableTriple<String, String, String>> subscribeTopics, SharedStrategyType strategyType) {
        SharedStrategy sharedStrategy = SharedStrategyFactory.createSharedStrategy(strategyType);
        String localNodeConsistentId = igniteCluster.localNode().consistentId().toString();
        ImmutableTriple<String, String, String> tr = sharedStrategy.searchSharedSubscribe(subscribeTopics, localNodeConsistentId);
        ClusterGroup group = getClusterNodeByConsistentId(tr.left);
        if (ObjectUtil.isNotEmpty(group)) {
            IgniteMessaging igniteMessaging = this.igniteIntegrate.getIgnite().message(group);
            clusterMessage.setTopic(tr.right);
            clusterMessage.setOriginTopic(tr.right);
            igniteMessaging.send(tr.right, clusterMessage);
        }
    }

    private ClusterGroup getClusterNodeByConsistentId(String consistentId) {
        ClusterNode node = igniteCluster
                .nodes()
                .stream()
                .filter(clusterNode -> StringUtils.equals(consistentId, clusterNode.consistentId().toString()))
                .findFirst().orElse(null);
        return Optional.ofNullable(node).isPresent() ? igniteCluster.forHost(node) : null;
    }

    @Override
    public Integrate getIntegrate() {
        return this.igniteIntegrate;
    }

}
